package org.pentaho.kettle.plugin.mqtt.steps.consumer;

import java.util.List;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.row.RowMeta;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.repository.Repository;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;
import org.pentaho.metastore.api.IMetaStore;

public class ConsumerStep extends BaseStep implements StepInterface {
    private static final Class<?> PKG = ConsumerStep.class;

    public ConsumerStep(StepMeta s, StepDataInterface stepDataInterface, int c, TransMeta t, Trans dis) {
        super(s, stepDataInterface, c, t, dis);
    }

    public boolean init(StepMetaInterface smi, StepDataInterface sdi) {
        ConsumerStepMeta meta = (ConsumerStepMeta)smi;
        ConsumerStepData data = (ConsumerStepData)sdi;
        if (super.init(meta, data)) {
            data.readsRows = this.getStepMeta().getRemoteInputSteps().size() > 0;
            List<StepMeta> previous = this.getTransMeta().findPreviousSteps(this.getStepMeta());
            if (previous != null && previous.size() > 0) {
                data.readsRows = true;
            }

            return true;
        } else {
            return false;
        }
    }

    public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
        ConsumerStepMeta meta = (ConsumerStepMeta)smi;
        ConsumerStepData data = (ConsumerStepData)sdi;
        Object[] r;
        if (data.readsRows) {
            r = this.getRow();
            if (r == null) {
                this.setOutputDone();
                return false;
            }

            if (this.first) {
                this.first = false;
                data.outputRowMeta = this.getInputRowMeta().clone();
                meta.getFields(data.outputRowMeta, this.getStepname(), (RowMetaInterface[])null, (StepMeta)null, this, (Repository)null, (IMetaStore)null);
            }
        } else {
            r = new Object[0];
            this.incrementLinesRead();
            if (this.first) {
                this.first = false;
                data.outputRowMeta = new RowMeta();
                meta.getFields(data.outputRowMeta, this.getStepname(), (RowMetaInterface[])null, (StepMeta)null, this, (Repository)null, (IMetaStore)null);
                MqttClientSubscriber client = new MqttClientSubscriber(meta, data, this, r);
                client.start();
            }
        }

        return true;
    }

    public void dispose(StepMetaInterface smi, StepDataInterface sdi) {
        ConsumerStepMeta meta = (ConsumerStepMeta)smi;
        ConsumerStepData data = (ConsumerStepData)sdi;
        super.dispose(meta, data);
    }
}
